//
// Created by LiuYou on 2021/8/19.
//

#ifndef HIGHPERFORMANCELINUXSERVERPROGRAMMING_LINUX2_PROCESSPOOL_HPP
#define HIGHPERFORMANCELINUXSERVERPROGRAMMING_LINUX2_PROCESSPOOL_HPP


#include <unistd.h>
#include <memory>
#include <sys/epoll.h>
#include <cassert>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/wait.h>


/**
 * @brief 描述一个子进程的类。
 * pid是目标子进程的PID, pipeFd是父子进程间通信用的管道。
 */
class Process {
public:
    Process();

    ~Process();

public:
    pid_t pid;
    int pipeFd[2];
};

/**
 * @brief 进程池类, 将它定义为模板类是为了代码复用。其模板参数是处理逻辑任务的类。
 */
template<typename T>
class ProcessPool {
public:
    /** 启动进程池。
     * <p>
     * 父进程中的idx值为-1, 子进程中idx值大于等于0, 我们据此判断接下来要运行的是父进程代码还是子进程代码。*/
    void run();

/** 将构造函数定义为私有, 因此我们只能通过后面的create()静态函数来创建ProcessPool实例。*/
    explicit ProcessPool(int listenFd);

private:
    /** 统一事件源。*/
    void setupSignalPipe();

    void runParent();

    void runChild();

public:
    /** 单例模式, 以保证程序最多创建一个ProcessPool实例, 这是程序正确处理信号的必要条件。*/
//    static std::unique_ptr<ProcessPool<T>> getInstance(int listenFd);
    static ProcessPool<T>* getInstance(int listenFd);

    ~ProcessPool() = default;

private:
    /** 进程池允许的最大子进程数量。*/
    static const int MAX_PROCESS_NUMBER = 16;

    /** 每个子进程最多能处理的客户数量。*/
    static const int USER_PER_PROCESS = 65536;

    /** epoll最多能处理的事件数。*/
    static const int MAX_EVENT_NUMBER = 10000;

    /** 进程池中的进程总数。*/
    static const int processNumber = 8;

    /** 子进程在池中的序号, 从0开始。*/
    int idx;

    /** 每个进程都有一个epoll内核事件表, 用 epollFd标识。*/
    int epollFd;

    /** 监听socket。*/
    int listenFd;

    /** 子进程通过 stop来决定是否停止运行。*/
    bool stop;

    /** 保存所有子进程的描述信息。*/
//    std::unique_ptr<Process> subProcess;
    std::array<std::unique_ptr<Process>, processNumber> subProcess;

    /** 进程池静态实例。*/
    // 如果使用 unique_ptr来实现单例模式。
//    /*static*/ std::unique_ptr<ProcessPool<T>> instance;

    // 放弃了使用 unique_ptr来实现单例模式。
    static ProcessPool<T>* instance;
};

template<typename T>
ProcessPool<T>* ProcessPool<T>::instance = nullptr;


///////////////////////////////////////////////////////////////////////////

/** 处于处理信号的管道, 以实现统一事件源。后面称之为信号管道。*/
static int signalPipeFd[2];


///////////////////////////////////////////////////////////////////////////

/*static*/ int setFdNonblocking(int fd);

/*static*/ void addFdToEpoll(int epollFd, int fd);

/*static*/ void addSignal(int sig, void(* signalHandler)(int), bool restart = true);

/*static*/ void signalHandler(int sig);

/**
 * @brief 从epollFd标识的epoll内核事件表中删除fd上的所有注册事件。
 * @param epollFd
 * @param fd
 */
/*static*/ void removeFd(int epollFd, int fd);


///////////////////////////////////////////////////////////////////////////

template<typename T>
ProcessPool<T>::ProcessPool(int listenFd):
        listenFd( listenFd ), idx( -1 ), stop( false ), epollFd() {
    assert((processNumber > 0) && (processNumber <= MAX_PROCESS_NUMBER));

    for (int i = 0; i < processNumber; ++i) {
        subProcess[i] = std::make_unique<Process>();
    }

    /* 创建processNumber个子进程, 并建立它们和父进程之间的管道。*/
    for (int i = 0; i < processNumber; ++i) {
        int ret = socketpair( PF_UNIX, SOCK_STREAM, 0, subProcess[i]->pipeFd );
        assert( ret == 0 );

        subProcess[i]->pid = fork();
        assert( subProcess[i]->pid >= 0 );
        if (subProcess[i]->pid > 0) {
            close( subProcess[i]->pipeFd[1] );
            continue;
        } else {
            close( subProcess[i]->pipeFd[0] );
            idx = i;
            break;
        }
    }
}

//template<typename T>
//std::unique_ptr<ProcessPool<T>>
//ProcessPool<T>::getInstance(int listenFd) {
///*    if (!instance.get()) {
//        instance = std::make_unique<ProcessPool<T>>( listenFd );
//    }
//    return instance;*/
//
//
//    return std::make_unique<ProcessPool<T>>( listenFd );
//}


template<typename T>
ProcessPool<T>*
ProcessPool<T>::getInstance(int listenFd) {
    if (!instance) {
        instance = new ProcessPool<T>( listenFd );
    }
    return instance;
}


template<typename T>
void ProcessPool<T>::run() {
    if (idx != -1) {
        runChild();
        return;
    }
    runParent();
}

template<typename T>
void ProcessPool<T>::setupSignalPipe() {
    /* 创建epoll事件监听表和信号管道。*/
    epollFd = epoll_create( 5 );
    assert( epollFd != -1 );

    socketpair( PF_UNIX, SOCK_STREAM, 0, signalPipeFd );

    setFdNonblocking( signalPipeFd[1] );
    addFdToEpoll( epollFd, signalPipeFd[0] );

    /* 设置信号处理函数。*/
    addSignal( SIGCHLD, signalHandler );
    addSignal( SIGTERM, signalHandler );
    addSignal( SIGINT, signalHandler );
    addSignal( SIGPIPE, SIG_IGN);
}

template<typename T>
void ProcessPool<T>::runParent() {
    setupSignalPipe();

    /* 父进程监听listenFd。*/
    addFdToEpoll( epollFd, listenFd );

    epoll_event epollEvents[MAX_EVENT_NUMBER];
    int subProcessCounter = 0;
    int newConnect = 1;

    while (!stop) {
        int readyFdNum = epoll_wait( epollFd, epollEvents, MAX_EVENT_NUMBER, -1 );
        if ((readyFdNum < 0) && (errno != EINTR)) {
            printf( "epoll failure\n" );
            break;
        }

        for (int i = 0; i < readyFdNum; ++i) {
            int fd = epollEvents[i].data.fd;
            if (fd == listenFd) {
                /* 如果有新连接到来, 就采用 Round Robin 方式将其分配给一个子进程处理。*/
                int i = subProcessCounter;
                do {
                    if (subProcess[i]->pid != -1) {
                        break;
                    }
                    i = (i + 1) % processNumber;
                } while (i != subProcessCounter);

                if (subProcess[i]->pid == -1) {
                    stop = true;
                    break;
                }
                subProcessCounter = (i + 1) % processNumber;
                send( subProcess[i]->pipeFd[0], (char*) &newConnect, sizeof( newConnect ), 0 );
                printf( "send request to child %d\n", i );
            }
                /* 下面处理父进程接收到的信号。*/
            else if ((fd == signalPipeFd[0]) && (epollEvents[i].events & EPOLLIN)) {
                char signals[1024];
                ssize_t rcvByteNum = recv( signalPipeFd[0], signals, sizeof( signals ), 0 );
                if (rcvByteNum <= 0) {
                    continue;
                } else {
                    for (int j = 0; j < rcvByteNum; ++j) {
                        switch (signals[j]) {
                            case SIGCHLD: {
                                pid_t pid;
                                int stat;
                                while ((pid = waitpid( -1, &stat, WNOHANG )) > 0) {
                                    for (int k = 0; k < processNumber; ++k) {
                                        /* 如果进程池中第i个子进程退出了, 则主进程关闭相应的通信管道, 并设置
                                         * 相应的pid为-1, 以标记该子进程已经退出。*/
                                        if (subProcess[k]->pid == pid) {
                                            printf( "child %d join\n", k );
                                            close( subProcess[k]->pipeFd[0] );
                                            subProcess[k]->pid = -1;
                                        }
                                    }
                                }
                                /* 如果所有子进程都已经退出了, 则父进程也退出。*/
                                stop = true;
                                for (int k = 0; k < processNumber; ++k) {
                                    if (subProcess[k]->pid != -1) {
                                        stop = false;
                                    }
                                }
                                break;
                            }
                            case SIGTERM: {
                                break;
                            }
                            case SIGINT: {
                                /* 如果父进程接受到终止信号, 那么就杀死所有子进程, 并等待它们全部结束。当然, 通知
                                 * 子进程结束更好的方法是向父子进程间的通信管道发送特殊数据, 读者不妨自己实现之。*/
                                printf( "kill all the child now\n" );
                                for (int k = 0; k < processNumber; ++k) {
                                    int pid = subProcess[k]->pid;
                                    if (pid != -1) {
                                        kill( pid, SIGTERM );
                                    }
                                }
                                break;
                            }
                            default: {
                                break;
                            }
                        }
                    }
                }
            } else {
                continue;
            }
        }
    }

    close( epollFd );
}

template<typename T>
void ProcessPool<T>::runChild() {
    setupSignalPipe();

    /* 每个子进程都通过其在进程池中的序号值idx找到与父进程通信的管道。*/
    int pipeFd = subProcess[idx]->pipeFd[1];
    /* 子进程需要监听管道文件描述符pipeFd, 因为父进程将通过它来通知子进程accept新连接。*/
    addFdToEpoll( epollFd, pipeFd );

    epoll_event epollEvents[MAX_EVENT_NUMBER];
    std::array<std::unique_ptr<T>, USER_PER_PROCESS> users;
    for (int i = 0; i < USER_PER_PROCESS; ++i) {
        users[i] = std::make_unique<T>();
    }

    while (!stop) {
        int readyFdNum = epoll_wait( epollFd, epollEvents, MAX_EVENT_NUMBER, -1 );
        if ((readyFdNum < 0) && (errno != EINTR)) {
            printf( "epoll failure\n" );
            break;
        }

        for (int i = 0; i < readyFdNum; ++i) {
            int fd = epollEvents[i].data.fd;
            if ((fd == pipeFd) && epollEvents[i].events & EPOLLIN) {
                int client = 0;
                /* 从父子进程间的管道读取数据, 并将结果保存在变量client中。如果读取成功, 则表示有辛苦胡连接到来。*/
                ssize_t rcvByteNum = recv( fd, (char*) &client, sizeof( client ), 0 );
                if (((rcvByteNum < 0) && (errno != EAGAIN)) || rcvByteNum == 0) {
                    continue;
                } else {
                    sockaddr_in clientAddress{};
                    socklen_t socklen = sizeof( clientAddress );
                    int connectFd = accept( listenFd, (sockaddr*) &clientAddress, &socklen );
                    if (connectFd < 0) {
                        perror( "accept()" );
                        continue;
                    } else {
                        addFdToEpoll( epollFd, connectFd );
                        /* 模板类T必须实现init()方法, 以初始化一个客户连接。我们直接使用connectFd来索引逻辑处理对象
                         * (T类型的对象), 以提高程序效率。*/
                        users[connectFd]->init( connectFd, <#initializer#>, <#initializer#>, <#initializer#>,
                                                clientAddress, epollFd );
                    }
                }
            } else if ((fd == signalPipeFd[0]) && (epollEvents[i].events & EPOLLIN)) {
                char signals[1024];
                ssize_t rcvByteNum = recv( signalPipeFd[0], signals, sizeof( signals ), 0 );
                if (readyFdNum <= 0) {
                    continue;
                } else {
                    for (int j = 0; j < rcvByteNum; ++j) {
                        switch (signals[j]) {
                            case SIGCHLD: {
                                pid_t pid;
                                int stat;
                                while ((pid = waitpid( -1, &stat, WNOHANG )) > 0) {
                                    continue;
                                }
                                break;
                            }
                            case SIGTERM: {
                                break;
                            }
                            case SIGINT: {
                                stop = true;
                                break;
                            }
                            default: {
                                break;
                            }
                        }
                    }
                }
            }
                /* 如果是其它可读数据, 那么必然是客户请求到来。调用逻辑处理对象的process方法处理之。*/
            else if (epollEvents[i].events & EPOLLIN) {
                users[fd]->process();
            } else {
                continue;
            }
        }
    }
    close( pipeFd );
    close( epollFd );
}


#endif //HIGHPERFORMANCELINUXSERVERPROGRAMMING_LINUX2_PROCESSPOOL_HPP
